package com.distributer.report.infrastructure.strom.topology;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

import com.distributer.report.infrastructure.strom.bolt.OrderStatisticsBolt;
import com.distributer.report.infrastructure.strom.bolt.ProductStatisticsBolt;
import com.distributer.report.infrastructure.strom.spout.OrderDataAnalysisSpout;


/**
 * 订单数据统计
 * 	1):实时计算产品的实时订单总金额，订单总数量。
 *  2):计算每60秒的前TOP5的保险产品。
 *  3):实时计算成交订单总金额。
 * @author xiaolin
 *
 */
public class OrderDataStatisticsTopology {
	
	public static void main(String[] args) {
		TopologyBuilder topologyBuilder = new TopologyBuilder();
		topologyBuilder.setSpout("orderDataAnalysisSpout", new OrderDataAnalysisSpout(), 1);
		topologyBuilder.setBolt("productStatisticsBolt", new ProductStatisticsBolt(), 2)
		.setNumTasks(4).fieldsGrouping("orderDataAnalysisSpout", new Fields("productCode"));
		topologyBuilder.setBolt("orderStatisticsBolt", new OrderStatisticsBolt(),1)
		.setNumTasks(1).shuffleGrouping("productStatisticsBolt");
		Config config = new Config();
		if (args != null && args.length > 0) {
			//集群模式
			config.setNumWorkers(3);
			try {
				StormSubmitter.submitTopology(args[0], config, topologyBuilder.createTopology());
			} catch (AlreadyAliveException e) {
				e.printStackTrace();
			} catch (InvalidTopologyException e) {
				e.printStackTrace();
			} catch (AuthorizationException e) {
				e.printStackTrace();
			}
		}else {
			//本地模式
			LocalCluster localCluster = new LocalCluster();
			localCluster.submitTopology("OrderDataStatisticsTopology", config, topologyBuilder.createTopology());
			try {
				Thread.sleep(10*1000*10);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			localCluster.killTopology("OrderDataStatisticsTopology");
			localCluster.shutdown();
		}
	}
}
